package com.atguigu.flink.window;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/12/14
 *
 *      性质:基于元素个数的窗口
 *                  countWindow
 *      计算： KeyBy
 *                  xxx。
 *                  每个key都有自己的窗口，各用各的！
 *      功能:  滑动
 *
 *
 */
public class Demo2_CountWindowKeyBy
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
                env
                    .socketTextStream("hadoop103", 8888)
                    .map(new WaterSensorMapFunction())
                    .keyBy(WaterSensor::getId)
                    //滚动窗口 size = slide = 3
                    //.countWindowAll(3)
                    //滑动，滑的小，重复计算
                    //.countWindowAll(3,1)
                    // //滑动，滑的小，漏算
                    .countWindow(3,1)
                    //ProcessWindowFunction<IN, OUT, KEY, W extends Window>
                    .process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>()
                    {
                        @Override
                        public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                           out.collect(key +"--" + MyUtil.parseIterable(elements));
                        }
                    }).setParallelism(10)
                    .print();
        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
